package com.beiding.jsonanalyzer.querylink;

import com.beiding.jsonanalyzer.analyze.map.BatchMapper;
import com.beiding.jsonanalyzer.analyze.map.Trigger;
import com.beiding.jsonanalyzer.flow.BranchNode;
import com.beiding.jsonanalyzer.flow.ValueNode;
import com.beiding.jsonanalyzer.flow.ValueNodeFactory;
import com.beiding.jsonanalyzer.flow.flowevents.BatchEndEvent;
import com.beiding.jsonanalyzer.flow.flowevents.BatchStartEvent;
import com.beiding.jsonanalyzer.flow.flowevents.DataNodeEvent;

import java.util.ArrayList;
import java.util.List;

/**
 * 聚集映射
 *
 * @author 丁常磊
 * @date 2022/6/11 10:51
 */

public class AggMapQueryLinkNode<W extends ValueNode> extends ComputeNode<W> {

    private BatchMapper<W> batchMapper;

    public AggMapQueryLinkNode(QueryLinkNode<W> pre, BatchMapper<W> batchMapper) {
        super(pre);
        this.batchMapper = batchMapper;
    }

    @Override
    protected void bindByComputeNode(BranchNode<W> preResult, BranchNode<W> newBranchNode) {

        ComputeNodeUtils.bindBaseEvent(preResult, newBranchNode);

        ValueNodeFactory<W> valueNodeFactory = newBranchNode.getBranch().getValueNodeFactory();

        //创建一个值发射器
        Trigger trigger = one -> {
            W w = valueNodeFactory.create();
            w.set(one);
            newBranchNode.signDataEvent(new DataNodeEvent<>(w));
        };

        //聚合批结果
        List<W> list = new ArrayList<>();

        //批开始不做处理
        preResult.onDataEvent(BatchStartEvent.eventType(), newBranchNode::signDataEvent);

        //来数据简单聚合
        preResult.onDataEvent(DataNodeEvent.eventType(), event -> {
            list.add(event.getData());
        });

        //批结束时处理
        preResult.onDataEvent(BatchEndEvent.eventType(), event -> {

            batchMapper.map(list, trigger);

            list.clear();
            newBranchNode.signDataEvent(event);
        });

    }
}
